package www.bwsensing.com.common.netty.handler;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import www.bwsensing.com.common.mqtt.MqttPushClient;
import www.bwsensing.com.common.utills.StringUtils;
import www.bwsensing.com.service.collection.database.dataobject.ServiceData;
import www.bwsensing.com.stream.DataStreamReceive;
import www.bwsensing.com.stream.object.DataMessage;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 朱永杰
 */
@Slf4j
@ChannelHandler.Sharable
public class NettyTcpServerHandler extends ChannelInboundHandlerAdapter implements DataPushHandle{
    private static final Integer MAX_WAIT_TIME =  4;
    /** 空闲次数 */
    private AtomicInteger idleCount = new AtomicInteger(1);

    private final ConcurrentHashMap<String,String> ipChannelCache = new ConcurrentHashMap<>();

    private static final ConcurrentHashMap<String, List<DataMessage>> IP_CACHE = new ConcurrentHashMap<>();
    /**
     * 服务配置信息
     */
    private final ServiceData serviceData;

    private final MqttPushClient mqttPushClient;

    public NettyTcpServerHandler(ServiceData serviceData, MqttPushClient mqttPushClient) {
        this.serviceData = serviceData;
        this.mqttPushClient = mqttPushClient;
    }

    /**
     * 建立连接时，发送一条消息
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("连接的客户端地址：{},客户端Id:{}",ctx.channel().remoteAddress(),ctx.channel().id().asShortText());
        ipChannelCache.put(ctx.channel().remoteAddress().toString(),ctx.channel().id().asShortText());
        IP_CACHE.put(ctx.channel().remoteAddress().toString(),new ArrayList<>());
        super.channelActive(ctx);
    }

    /**
     * 超时处理 如果5秒没有接受客户端的心跳，就触发; 如果超过两次，则直接关闭;
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            // 如果读通道处于空闲状态，说明没有接收到心跳命令
            if (IdleState.READER_IDLE.equals(event.state())) {
                log.info("心跳超时,IP地址:{}",ctx.channel().remoteAddress());
                if (idleCount.get()>MAX_WAIT_TIME) {
                    log.info("心跳重试后再超时,IP地址:{},关闭通道",ctx.channel().remoteAddress());
                    ctx.channel().close();
                    ipChannelCache.remove(ctx.channel().remoteAddress().toString());
                }
                idleCount.addAndGet(1);
            }
        }
    }

    /**
     * 业务逻辑处理
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        long currentTime = System.currentTimeMillis();
        Timestamp timestamp = new Timestamp(currentTime);
        try {
            String rawText = msg.toString().replace("\r","").replace(" ","");;
            if (StringUtils.isNotEmpty(rawText)){
                String ipAddress = ctx.channel().remoteAddress().toString();
                if (StringUtils.isEmpty(IP_CACHE.get(ipAddress))){
                    List<DataMessage> messages = new ArrayList<>();
                    messages.add(new DataMessage(timestamp,rawText));
                    IP_CACHE.put(ipAddress,messages);
                } else {
                    IP_CACHE.get(ipAddress).add(new DataMessage(timestamp,rawText));
                }
                log.info("Data receive Client address:{}，Raw message:{}", ipAddress,msg);
            }
        } catch (Exception e) {
            log.warn("netty channel Read error Message:{}",e.getLocalizedMessage());
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }


    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        //客户端断开后批量保存
        String ipAddress = ctx.channel().remoteAddress().toString();
        if (IP_CACHE.get(ipAddress).size()>0){
            DataStreamReceive receiveData = getDataStreamPush(IP_CACHE.get(ipAddress),"TCP",serviceData);
            mqttPushClient.publishEvent(receiveData);
            log.info(receiveData.getMessage());
            log.info("facility data pushed,Client address:{}",ipAddress);
        }
        IP_CACHE.remove(ipAddress);
        log.info("link down ,Client address:{}",ctx.channel().remoteAddress());
    }

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.warn("TCP Receive error:{}",cause.getLocalizedMessage());
        ctx.close();
    }
}
